package com.mq.mqserver.datacenter;

import com.mq.common.exception.MqException;
import com.mq.mqserver.core.Binding;
import com.mq.mqserver.core.Exchange;
import com.mq.mqserver.core.MSGQueue;
import com.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/*
* 使用和各类统一管理内存中所有数据
* */
public class MemoryDataCenter {
    //key是 exchangeName , value 是exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //key是 queueName , value 是 queue 对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //第一个key是 exchangeName , 第二个 key 是 queueName ,value 是 Binding 对象
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingExchangeNameMap = new ConcurrentHashMap<>();
    //key是 messageName , value 是 Message 对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //key是 queueName , value 是 Message 链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //第一个key是 queueName , 第二个key是messageId , value 是 Message 对象,保存已读未回的消息
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功! exchangeName = " + exchange.getName());
    }

    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }
    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        if(getExchange(exchangeName) ==null) {
            System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName = " + exchangeName);
        }else{
            System.out.println("[MemoryDataCenter] 交换机删除失败! exchangeName = " + exchangeName);
        }
    }
    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功! queueName = " + queue.getName());
    }
    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }
    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 新队列删除成功! queueName = " + queueName);
    }
    public void insertBindingExchangeName(Binding binding) throws MqException {
        //先使用 exchangeName 查一下对应的哈希表是否存在，不存在就创建一个
        ConcurrentHashMap<String, Binding> bindingQueueNameMap = bindingExchangeNameMap.get(binding.getExchangeName());
        if (bindingQueueNameMap == null) {
            bindingQueueNameMap = new ConcurrentHashMap<>();
            bindingExchangeNameMap.put(binding.getExchangeName(), bindingQueueNameMap);
        }

//        ConcurrentHashMap<String,Binding> bindingQueueNameMap =
//                bindingExchangeNameMap.computeIfAbsent(binding.getExchangeName(),k -> new ConcurrentHashMap<>());
//
//    }
        //再根据 queueName 查一下对应的 binding 值，要是存在就说明 exchange 与 queue 已经绑定，抛出异常
        synchronized (bindingQueueNameMap) {
            //此处虽然是 ConcurrentHashMap ，但是涉及到get和put两步操作，使用synchronized将两步操作合并成一个原子操作
            if (bindingQueueNameMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在! exchangeName = " + binding.getExchangeName()
                        + ",queueName = " + binding.getQueueName());
            }
            bindingQueueNameMap.put(binding.getQueueName(), binding);
            System.out.println("[MemoryDataCenter] 新绑定添加成功! exchangeName = "+binding.getExchangeName()+
                    "queueName = " + binding.getQueueName());
        }
    }

    //获取绑定写两个版本
    //1.根据 exchangeName 和 queueName 确定唯一一个 Binding
    //2.根据 exchangeName 获取所有 Binding
    public Binding getBinding(String exchangeName,String queueName){
        //要是 exchangeName 对应的哈希表非空，返回空值，否侧返回哈希表中 queueName 对应的binding
        ConcurrentHashMap<String,Binding> bindingQueueNameMap = bindingExchangeNameMap.get(exchangeName);
        if(bindingQueueNameMap == null){
            return null;
        }

        return bindingQueueNameMap.get(queueName);
        //这里没有使用 synchronized 加锁是因为，两个get操作都是获取信息，最坏的情况就是没有获取到信息，再次获取信息就行

    }
    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName){
        return bindingExchangeNameMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String,Binding> bindingQueueNameMap = bindingExchangeNameMap.get(binding.getExchangeName());
        if(bindingQueueNameMap == null){
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName = " +
                    binding.getExchangeName() + ",queueName = " + binding.getQueueName());
        }

        bindingQueueNameMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName = "+
                binding.getExchangeName()+ "queueName = " + binding.getQueueName());
    }

    //添加消息
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新信息添加成功! messageId = "+
                message.getMessageId());
    }

    //根据id删除消息
    public Message getMessage(String messageId){
        return messageMap.get(messageId);

    }

    public void deleteMessage(String messageId){
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 删除消息成功! messageId = "+messageId);
    }
    //发送消息到指定队列
    public void sendMessage(MSGQueue queue,Message message){
        //先查找queue对应的消息链表，要是没有链表，就创建一个新的
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent
                (queue.getName(),k-> new LinkedList<>());
        //再把数据加到messages链表里面
        synchronized (messages) {
            messages.add(message);
            //在这里把消息也往
        }
        //这里重复插入也没关系，只要是确保 messageId 能够查到 message 一致
        addMessage(message);
        System.out.println("[MemoryDataCenter] 信息投递到队列中! messageId = "+
                message.getMessageId() + "queueName = " +queue.getName());

    }
    //从队列中获取消息
    public Message pollMessage(String queueName) {
        //消息是以链表的形式，表示队列存储的，从Map中根据队列名获取消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            return null;
        }
        synchronized (messages) {
            if (messages.size() == 0) {
                return null;
            }
            //由于是队列，先进先出，取出链表中元素，就进行头删，返回取出消息之后的队列链表
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中删除! messageId = " +
                    currentMessage);
            return currentMessage;
        }
    }
    //获取指定队列中的消息个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }
    //添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message){
        ConcurrentHashMap<String,Message> messageHashMap =
                queueMessageWaitAckMap.computeIfAbsent(queueName,k->new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列! messageId = " +
                message.getMessageId());
    }
    //删除未确认的消息（已经确认了）
    public void removeMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if(messageHashMap == null){
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列中删除! messageId = " +
                messageId);
    }
    //获取指定的未确认消息
    public Message getMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if(messageHashMap == null){
            return null;
        }
        return messageHashMap.get(messageId);
    }

//从硬盘上读取数据，把硬盘中之前持久化存储的各个维度数据恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //0.清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingExchangeNameMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        //1.恢复所有的交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchange();
        for(Exchange exchange : exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }
        //2.恢复所有的队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueue();
        for(MSGQueue queue : queues){
            queueMap.put(queue.getName(),queue);
        }
        //3.恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBinding();
        for(Binding binding : bindings){
            ConcurrentHashMap<String,Binding> bindingMap =
                    bindingExchangeNameMap.computeIfAbsent(binding.getExchangeName(),k->new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }
        //4.恢复所有的消息数据
        //遍历所有的队列，根据每个队列的名字获取所有的消息
        for(MSGQueue queue : queues){
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for(Message message :messages){
                messageMap.put(message.getMessageId(),message);
            }
        }
        //针对未确认的消息，这部分内存中的数据不需要从磁盘中恢复，之前考虑存硬盘储的时候，也没有设定这一块
        //一旦在等待ack的过程中，服务器重启了，此时这些未被确认的数据，就恢复成未被取走的消息
        //这个消息在硬盘上存储的时候，就是当作“未被取走”
    }

}
